package com.afdor.rws.event.channel.redis;

import com.afdor.rws.annotation.extension.Extension;
import com.afdor.rws.event.Event;
import com.afdor.rws.event.channel.client.consumer.AbstractEventConsumer;
import com.afdor.rws.event.channel.client.consumer.EventConsumer;
import com.afdor.rws.event.channel.client.consumer.handler.EventRetryStrategy;
import lombok.EqualsAndHashCode;
import lombok.Getter;
import lombok.NonNull;
import lombok.Setter;
import lombok.ToString;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.ListOperations;
import org.springframework.data.redis.core.RedisTemplate;

import java.util.List;

/**
 * 基于redis的事件消费者实现
 *
 * @author 悭梵
 * @date Created in 2018-08-17 10:24
 */
@Slf4j
@ToString
@EqualsAndHashCode
@Extension(value = "redis", order = Extension.NORMAL_ORDER - 100, scope = Extension.Scope.PROTOTYPE, interfaceClass = EventConsumer.class)
public class RedisEventConsumer extends AbstractEventConsumer {

    @Getter
    @Setter
    private RedisTemplate redisTemplate;

    private ListOperations listOperations;

    private String queueName, retryQueueName;

    public RedisEventConsumer(@NonNull RedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
    }


    @Override
    protected void doStart() {
        super.doStart();
        buildQueueName(getEventSourceName());
        this.listOperations = this.redisTemplate.opsForList();
    }

    @Override
    protected boolean consume() {
        RedisConnection redisConnection = this.redisTemplate.getRequiredConnectionFactory().getConnection();
        List<byte[]> results = redisConnection.bLPop(10, this.redisTemplate.getKeySerializer().serialize(this.retryQueueName), this.redisTemplate.getKeySerializer().serialize(this.queueName));
        if (results == null || results.isEmpty()) {
            return Boolean.FALSE;
        }
        String messageBody = (String) this.redisTemplate.getValueSerializer().deserialize(results.get(1));
        Event event = getSerialization().unserialize(messageBody);
        if (event == null) {
            return Boolean.FALSE;
        }
        try {
            if (log.isInfoEnabled()) {
                log.info("<===== 收到消息：" + event);
            }
            getEventHandler().handler(event);
        } catch (Exception e) {
            EventRetryStrategy eventRetryStrategy = getEventRetryStrategy();
            if (eventRetryStrategy != null) {
                eventRetryStrategy.retry(event);
            }
            log.error("事件处理异常", e);
        } finally {
            redisConnection.close();
        }
        return Boolean.TRUE;
    }

    /**
     * 获取队列名称
     *
     * @param eventSourceName
     * @return
     */
    private String buildQueueName(String eventSourceName) {
        if (queueName == null) {
            synchronized (this) {
                if (queueName == null) {
                    queueName = RedisUtils.getQueueName(eventSourceName);
                    retryQueueName = RedisUtils.getRetryQueueName(eventSourceName);
                }
            }
        }
        return queueName;
    }

}
